#pragma once

/*
bug
1.send返回值为0时当错误处理了
2.socktefd关闭时没有置为-1，导致重复关闭，或者其他线程打开就被另一个线程关闭了 *****
3.it == nullptr -> it == .end()
4.连接状态没有设置好
5.buffer的拷贝的长度少加了
6.需要注意一些函数在RunInLoop中，一些在QueuePush中
*/

#include <iostream>
#include <string>
#include <cstring>
#include <cassert>
#include <vector>
#include <sys/epoll.h>
#include <functional>
#include <unordered_map>
#include <sys/eventfd.h>
#include <memory>
#include <functional>
#include <thread>
#include <fcntl.h>
#include <sys/socket.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <typeinfo>
#include <mutex>
#include <condition_variable>
#include <ctime>
#include <sys/timerfd.h>
#include <sys/select.h>
#include <signal.h>
#include "Log.hpp"

using namespace LogMudule;

using EVENT_CALLBACK = std::function<void()>;
using TaskCallback = std::function<void()>;
using CleanCallback = std::function<void()>;
using Functor = std::function<void()>;

const int FD_DEFAULT = -1;
const int MAX_EVENT = 1024;
const int BUFFER_DEFAULT_SIZE = 1024;
const int SOCKFD_DEFAULT = -1;
const int LISTEN_BLOCK_DEFAULT = 1024;
const int CAP_DEFAULT = 60;

class EventLoop;
class Poller;

class Buffer
{
public:
    Buffer()
        : _read_index(0),
          _write_index(0),
          _buffer(BUFFER_DEFAULT_SIZE)
    {
    }

    // 获取_buffer首位地址
    char *Begin()
    {
        return (char *)(&(*_buffer.begin()));
    }

    // 获取可读位置的地址
    char *ReadPtr()
    {
        return Begin() + _read_index;
    }

    // 获取可写位置的地址
    char *WritePtr()
    {
        return Begin() + _write_index;
    }

    // 可读区域大小
    int64_t ReadSize()
    {
        return _write_index - _read_index;
    }

    // 头部可写区域大小
    int64_t HeadWriteSize()
    {
        return _read_index;
    }

    // 尾部可写区域大小
    int64_t TailWriteSize()
    {
        return _buffer.size() - _write_index;
    }

    // 读指针进行偏移
    void ReadIndexSkewing(uint64_t len)
    {
        if(len == 0)    return;     //111
        assert(len <= ReadSize());
        _read_index += len;
    }

    // 写指针进行偏移
    void WriteIndexSkewing(uint64_t len)
    {
        assert(len <= TailWriteSize());
        _write_index += len;
    }

    // 保证有足够空间插入
    void GuaranteeSpaceEnough(uint64_t len)
    {
        // 1.尾部有足够空间
        if (len <= TailWriteSize())
        {
            return;
        }
        // 2.头部+尾部有足够空间
        if (len <= HeadWriteSize() + TailWriteSize())   //111
        {
            uint64_t readsize = ReadSize();
            std::copy(ReadPtr(), WritePtr()+ readsize, Begin()); //111
            _read_index = 0;
            _write_index = readsize;
        }
        // 3.需要扩容
        else
        {
            _buffer.resize(_write_index + len); ///111
        }
    }

    // 进行写入
    void Write(const void *in, uint64_t len)
    {
        if (len == 0)
        {
            return;
        }

        GuaranteeSpaceEnough(len);
        const char *temp_in = (const char *)in;
        std::copy(temp_in, temp_in + len, WritePtr());
    }

    // 进行写入并偏移
    void WritePush(const void *in, uint64_t len)
    {
        Write(in, len);
        WriteIndexSkewing(len);
    }

    // 进行写入 - string
    void WriteString(const std::string &in)
    {
        Write(in.c_str(), in.size());
    }

    // 进行写入并偏移 - string
    void WriteStringPush(const std::string &in)
    {
        Write(in.c_str(), in.size());
        WriteIndexSkewing(in.size());
    }

    // 进行写入 - Buffer
    void WriteBuffer(Buffer &in)
    {
        Write(in.ReadPtr(), in.ReadSize());
    }

    // 进行写入并偏移 - Buffer
    void WriteBufferPush(Buffer &in)
    {
        WriteBuffer(in);
        WriteIndexSkewing(in.ReadSize());
    }

    // 进行读取
    void Read(void *out, uint64_t len)
    {
        assert(len <= ReadSize());
        char *read_ptr = ReadPtr();
        std::copy(read_ptr, read_ptr + len, (char *)out);
    }

    // 进行读取并偏移
    void ReadPop(void *out, uint64_t len)
    {
        Read(out, len);
        ReadIndexSkewing(len);
    }

    // 进行读取 - string
    std::string ReadString(uint64_t len)
    {
        if (len == 0)
            return "";
        std::string ret;
        ret.resize(len);

        Read(&ret[0], len);

        return ret;
    }

    // 进行读取并偏移
    std::string ReadStringPop(uint64_t len)
    {
        std::string ret = ReadString(len);
        ReadIndexSkewing(len);
        return ret;
    }

    // 进行行读取
    std::string GetLine()
    {
        char *res = FindLine();
        if (res == nullptr)
            return "";

        return ReadString(res - ReadPtr() + 1);
    }

    // 进行行读取并进行偏移
    std::string GetLinePop()
    {
        std::string res = GetLine();
        ReadIndexSkewing(res.size());
        return res;
    }

    // 清空缓冲区
    void Clear()
    {
        _read_index = 0;
        _write_index = 0;
    }

    ~Buffer() = default;

private:
    char *FindLine()
    {
        char *ret = (char *)memchr(ReadPtr(), '\n', ReadSize());
        return ret;
    }

private:
    std::vector<char> _buffer;
    uint64_t _read_index;
    uint64_t _write_index;
};

class Socket
{
public:
    Socket()
        : _sockfd(SOCKFD_DEFAULT)
    {
    }

    Socket(int sockfd,bool flag = false)
        : _sockfd(sockfd)
    {
        if(flag)    //XZY
        {
            SockBlock();
        }
    }

    ~Socket()
    {
        Close();
    }

    // 创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            LOG(LogLevel::ERROR) << "create sockfd errof ...";
            return false;
        }
        return true;
    }

    // 绑定窗口
    bool Bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);

        int n = bind(_sockfd, (struct sockaddr *)&addr, len);
        if (n < 0)
        {
            LOG(LogLevel::ERROR) << "bind error ...";
            return false;
        }

        return true;
    }

    // 设置为监听状态
    bool Listen(int block = LISTEN_BLOCK_DEFAULT)
    {
        int n = listen(_sockfd, block);
        if (n < 0)
        {
            LOG(LogLevel::ERROR) << "listen erron...";
            return false;
        }

        return true;
    }

    // 连接服务端
    bool Connect(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);

        int n = connect(_sockfd, (struct sockaddr *)&addr, len);
        if (n < 0)
        {
            LOG(LogLevel::ERROR) << "connect error ...";
            return false;
        }

        return true;
    }

    // 获取新连接
    int Accept()
    {
        int new_sockfd = accept(_sockfd, nullptr, nullptr);
        if (new_sockfd < 0) //XZY
        {
            if(errno == EAGAIN || errno == EWOULDBLOCK)
            {
                return -1;
            }
            else if(errno == EINTR)
            {
                return -2;
            }

            LOG(LogLevel::ERROR) << "accept error ...";
            return -3;
        }

        return new_sockfd;
    }

    // 读取数据
    ssize_t Recv(void *buf, size_t len, int flag = 0)
    {
        ssize_t size = recv(_sockfd, buf, len, flag);
        if (size <= 0)
        {
            if (errno == EAGAIN  || errno == EWOULDBLOCK)
            {
                return 0;
            }
            else if(errno == EINTR)
            {
                return -2;
            }
            else if(size == 0)
            {
                LOG(LogLevel::INFO) << "对端关闭";
                return -1;
            }
            LOG(LogLevel::ERROR) << "recv error ...: " << size;
            return -1;
        }

        return size;
    }

    // 非阻塞读取数据
    ssize_t NotBlockRecv(void *buf, size_t len)
    {
        return Recv(buf, len, MSG_DONTWAIT);
    }

    // 写数据
    ssize_t Send(const void *buf, size_t len, int flag = 0)
    {
        ssize_t size = send(_sockfd, buf, len, flag);
        if (size < 0)
        {
            if (errno == EAGAIN  || errno == EWOULDBLOCK)
            {
                return 0;
            }
            else if(errno == EINTR)
            {
                return -2;
            }
            LOG(LogLevel::ERROR) << "send error ...";
            return -1;
        }
        else if(size == 0)
        {
            return -1;
        }

        return size;
    }

    // 非阻塞写数据
    /*关于非阻塞读写：如果将套接字设置为非阻塞读写，无论是否设置flag都默认为非阻塞读写，如果不设置就需要看flag设置为什么*/
    ssize_t NotBlockSend(const void *buf, size_t len)
    {
        if(len == 0) return 0; //111
        return Send(buf, len, MSG_DONTWAIT);
    }

    // 设置地址重用
    void ReusingAddr()
    {
        int optval = 1;
        int n = setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
        if (n < 0)
        {
            LOG(LogLevel::ERROR) << "setsockopt addr  error ...";
        }

        optval = 1;
        n = setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval));
        if (n < 0)
        {
            LOG(LogLevel::ERROR) << "setsockopt port  error ...";
        }
    }

    // 设置为套接字非阻塞
    void SockBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL,0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }

    // 创建服务端套接字
    bool CreateServer(const std::string &ip, uint16_t port, bool flag = false)
    {
        if (Create() == false)
            return false;
        //启动地址重用
        ReusingAddr();

        if (Bind(ip, port) == false)
            return false;

        if (Listen() == false)
            return false;

        if (flag)
        {
            SockBlock();
        }

        LOG(LogLevel::INFO) << "创建服务端监听套接字完成";

        return true;
    }

    // 创建客户端套接字
    bool CreateClient(const std::string &ip, uint16_t port, bool flag = false)
    {
        if (Create() == false)
            return false;
        if (Connect(ip, port) == false)
            return false;

        LOG(LogLevel::INFO) << "创建客户端套接字完成";
        return true;
    }

    // 关闭连接
    void Close()
    {
        if(_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }  
    }

    // 获取sockfd
    int Fd()
    {
        return _sockfd;
    }

private:
    int _sockfd;
};

class Channel
{
public:
    Channel(int fd, EventLoop *loop)
        : _fd(fd),
          _loop(loop),
          _event(0),
          _ready_event(0)
    {}

    // 设置相关的回调函数
    void SetReadCallback(const EVENT_CALLBACK &cb)
    {
        _read_callback = cb;
    }
    void SetWriteCallback(const EVENT_CALLBACK &cb)
    {
        _write_callback = cb;
    }
    void SetCloseCallback(const EVENT_CALLBACK &cb)
    {
        _close_callback = cb;
    }
    void SetErrorCallback(const EVENT_CALLBACK &cb)
    {
        _error_callback = cb;
    }
    void SetEventCallback(const EVENT_CALLBACK &cb)
    {
        _event_callback = cb;
    }

    // 是否监控读事件
    bool IsMonitorRead()
    {
        return (_event & EPOLLIN);
    }

    // 是否监控写事件
    bool IsMonitorWrite()
    {
        return (_event & EPOLLOUT);
    }

    // 启用读事件
    void EnableRead()
    {
        //_event |= EPOLLIN; 
        _event |= (EPOLLIN | EPOLLET);  //XZY
        Updata();
    }

    // 启用写事件
    void EnableWrite()
    {
        _event |= (EPOLLOUT | EPOLLET); //XZY
        Updata();
    }

    // 解除读事件监控
    void RelieveRead()
    {
        _event &= (~EPOLLIN);
        Updata();
    }

    // 解除写事件监控
    void RelieveWrite()
    {
        _event &= (~EPOLLOUT);
        Updata();
    }

    // 解除所有事件的监控
    void RelieveAll()
    {
        _event = 0;
        Updata();
    }

    // 设置就绪事件
    void SetEvent(int event)
    {
        _ready_event = event;
    }

    // 添加或者修改监控
    void Updata();

    // 删除监控
    void Remove();

    // 处理函数
    void EventHandle()
    {
        // 可读事件就绪
        if (_ready_event & EPOLLIN || _ready_event & EPOLLRDHUP || _ready_event & EPOLLPRI)
        {
            if (_read_callback)
            {
                _read_callback();
            }
        }

        /*
            读事件可能导致连接关闭，但是关闭这件事是放到EventLoop的任务队列中完成的，
            所以不会因为连接被销毁而导致下面的函数执行出错
        */

        // 可写事件就绪
        if (_ready_event & EPOLLOUT)
        {
            if (_write_callback)
            {
                _write_callback();
            }
        }
        // 关闭事件就绪
        else if (_ready_event & EPOLLHUP)
        {
            if (_close_callback)
            {
                _close_callback();
            }
        }
        // 错误事件就绪
        else if (_ready_event & EPOLLERR)
        {
            if (_error_callback)
            {
                _error_callback();
            }
        }

        // 任意事件
        if (_event_callback)
        {
            _event_callback();
        }
    }

    // 返回Fd
    int Fd()
    {
        return _fd;
    }

    // 返回event
    uint32_t GetEvent()
    {
        return _event;
    }

private:
    uint32_t _event;        //监控的事件
    uint32_t _ready_event;  //监控触发的事件
    int _fd = FD_DEFAULT;   //文件描述符
    EventLoop *_loop;       

    //相关函数
    EVENT_CALLBACK _read_callback = nullptr;    //读回调
    EVENT_CALLBACK _write_callback = nullptr;   //写回调
    EVENT_CALLBACK _close_callback = nullptr;   //关闭回调
    EVENT_CALLBACK _error_callback = nullptr;   //错误回调
    EVENT_CALLBACK _event_callback = nullptr;   //任意事件回调
};


//监控类
/*修改*/
class Poller
{
public:
    Poller()
    {
        _epfd = epoll_create(MAX_EVENT);
        if (_epfd < 0)
        {
            LOG(LogLevel::ERROR) << " epoll_create error ... ";
            abort();
        }
        LOG(LogLevel::INFO) << "创建epoll成功 : " << _epfd;
    }

    // 修改或者添加监控
    void UpdataEvent(Channel *channel)
    {
        int fd = channel->Fd();
        bool ret = FideEvent(fd);
        if (ret == false)
        {
            _fd_channel.insert(std::make_pair(fd,channel));
            Updata(channel, EPOLL_CTL_ADD);
            return;
        }

        Updata(channel, EPOLL_CTL_MOD);
    }

    // 删除监控
    void RemoveEvent(Channel *channel)
    {
        auto it = _fd_channel.find(channel->Fd());
        if(it != _fd_channel.end())
        {
            
            _fd_channel.erase(it);
            Updata(channel, EPOLL_CTL_DEL);
        }
    }

    // 开始监控
    void Poll(std::vector<Channel *> *out_event)
    {
        int n = epoll_wait(_epfd, _evs, MAX_EVENT, -1);
        if (n < 0)
        {
            if (errno == EINTR) {
                return ;
            }
            LOG(LogLevel::ERROR) << "监控出错";
            abort();
        }

        for (int i = 0; i < n; ++i)
        {
            auto it = _fd_channel.find(_evs[i].data.fd);
            if (it == _fd_channel.end())
            {
                LOG(LogLevel::ERROR) << "channel 不存在 _fd_channel 中";
                abort();
            }
            it->second->SetEvent(_evs[i].events);
            out_event->push_back(it->second);
        }
    }

private:
    bool FideEvent(int fd)
    {
        auto it = _fd_channel.find(fd);
        if (it == _fd_channel.end())
        {
            return false;
        }

        return true;
    }

    void Updata(Channel *channel, int op)
    {
        struct epoll_event ev;
        ev.data.fd = channel->Fd();
        ev.events = channel->GetEvent();

        if(fcntl(channel->Fd(), F_GETFD) == -1)
        {
            LOG(LogLevel::ERROR) << " 文件描述符被关 :  " << channel->Fd(); 
        }
        
        int n = epoll_ctl(_epfd, op, channel->Fd(), &ev);
        if (n < 0)
        {
            if(errno == EBADF)
            {
                LOG(LogLevel::ERROR) << " 文件描述符被关 :  " << _epfd;
            }
            if( errno == EEXIST) 
            {
                LOG(LogLevel::ERROR) << " 已经存在 " << n;
            }
            LOG(LogLevel::ERROR) << "epoll_ctl error ... n: " << n;
        }
        return;
    }

private:
    int _epfd;                                      //eventfd
    struct epoll_event _evs[MAX_EVENT];             //获取监控完成的事件
    std::unordered_map<int, Channel *> _fd_channel; //管理文件描述符和Channel
};


//定时相关类
/*通过创建对象时创建任务析构对象时执行任务*/
class TaskClock
{
public:
    TaskClock(uint64_t id, uint32_t interval, TaskCallback tc)
        : _id(id),
          _interval(interval),
          _tc(tc),
          _switch(true)
    {}

    // 返回间隔时间
    uint32_t GetInterval()
    {
        return _interval;
    }

    // 设置清理函数
    void SetCleanCallback(CleanCallback cc)
    {
        _cc = cc;
    }

    // 关闭时钟
    void ShutTaskClock()
    {
        _switch = false;
    }

    // 释放
    ~TaskClock()
    {
        LOG(LogLevel::DEBUG)<<"释放定时任务";
        if (true == _switch)
        {
            _tc();
        }
        _cc();
    }

private:
    uint64_t _id;       // 时钟id
    uint32_t _interval; // 间隔时间
    bool _switch;       // 是否执行函数
    TaskCallback _tc;   // 时间到了，触发的函数
    CleanCallback _cc;  // 删除时钟对象的函数
};


class TimerWheel
{
    using WeatPtr = std::weak_ptr<TaskClock>;
    using TaskPtr = std::shared_ptr<TaskClock>;

public:
    TimerWheel(EventLoop *loop, int cap = CAP_DEFAULT)
        : _tick(0),
          _cap(cap),
          _tasks(_cap),
          _timerfd(CreateTimerfd()),
          _loop(loop),
          _timerchannel(std::make_unique<Channel>(_timerfd, _loop))
    {
        _timerchannel->SetReadCallback(std::bind(&TimerWheel::RunTask, this));
        _timerchannel->EnableRead();
    }

    //添加一个时钟
    void TimerAdd(uint64_t id, uint32_t interval, TaskCallback tc);

    //刷新时钟
    void RefreshTask(uint64_t id);

    //下线一个时钟
    void ShutTaskClock(uint64_t id);

    /*该接口存在线程安全，只能给Loop线程使用*/
    bool IsTimer(uint64_t id)
    {
        auto it = _id_task.find(id);
        if (it == _id_task.end()) //111
        {
            return false;
        }
        return true;
    }

    ~TimerWheel()
    {}

private:
    //创建一个定时器
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0) //111
        {
            LOG(LogLevel::ERROR) << "timerfd_create error";
            abort();
        }

        struct itimerspec iterval;
        iterval.it_value.tv_sec = 1;
        iterval.it_value.tv_nsec = 0;
        iterval.it_interval.tv_sec = 1;
        iterval.it_interval.tv_nsec = 0;

        timerfd_settime(timerfd, 0, &iterval, nullptr);
        return timerfd;
    }

    //读取定时器fd
    int ReadTimerfd()
    {
        int64_t exp;
        int n = read(_timerfd, &exp, sizeof(exp));
        if (n < 0)
        {
            LOG(LogLevel::ERROR) << "read timerfd error";
            abort();
        }
        return exp;
    }

    //执行定时任务
    void RunTask()
    {
        int count  = ReadTimerfd();

        /*
        因为时钟每一秒发送一次，每一次要跳一格，但是有些任务可能会耗时较长，
        所以过了几次就应该执行几格
        */
        for(int i = 0; i < count; ++i)
        {
            RunOne();
        }
            
    }

    //执行一次定时任务
    void RunOne()
    {
        _tick = (_tick + 1) % _cap;
        _tasks[_tick].clear();
    }

    //添加定时任务
    void TimerAddInLoop(uint64_t id, uint32_t interval, TaskCallback tc)
    {
       // TaskPtr taskptr = std::make_shared<TaskClock>(id, interval, tc);
       LOG(LogLevel::DEBUG)<<"interval : " <<interval;
       TaskPtr taskptr (new TaskClock(id, interval, tc));
        taskptr->SetCleanCallback(std::bind(&TimerWheel::Remove, this, id));
        int pos = (interval + _tick) % _cap;
        _tasks[pos].push_back(taskptr);
        _id_task[id] = WeatPtr(taskptr);
    }

    //刷新定时任务
    void RefreshTaskInLoop(uint64_t id)
    {
        auto it = _id_task.find(id);
        if (it == _id_task.end())
        {
            return;
        }

        TaskPtr taskptr = it->second.lock();
        int pos = (taskptr->GetInterval() + _tick) % _cap;
        _tasks[pos].push_back(taskptr);
    }

    //定时任务下线
    void ShutTaskClockInLoop(uint64_t id)
    {
        auto it = _id_task.find(id);
        if (it == _id_task.end())
        {
            return;
        }

        TaskPtr taskptr = it->second.lock();
        if(taskptr != nullptr)
        {
            taskptr->ShutTaskClock();
        }
    }

    //删除一个定时任务
    void Remove(uint64_t id)
    {
        auto it = _id_task.find(id);
        if (it == _id_task.end())
        {
            return;
        }
        _id_task.erase(it);
    }

private:
    int _tick;  //下标
    int _cap;   //长度
    std::vector<std::vector<TaskPtr>> _tasks;   
    std::unordered_map<uint64_t, WeatPtr> _id_task;
    int _timerfd;   //定时fd
    EventLoop *_loop;   
    std::unique_ptr<Channel> _timerchannel;
};


class EventLoop
{
public:
    EventLoop()
        : _event_fd(CreateEventfd()),
          _channel(new Channel(_event_fd, this)),
          _thread_id(std::this_thread::get_id()),
          _timer_wheel(this)
    {
        _channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
        _channel->EnableRead();
    }

    // 把任务队列的任务全部执行了
    void ExecuteTaskAll()
    {
        std::vector<Functor> replace_queue;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.swap(replace_queue); //111
            //replace_queue.swap(_tasks);
        }

        for (auto &e : replace_queue)
        {
            e();
        }
    }

    // 是EventLoop线程
    bool IsEventLoop()
    {
        return (_thread_id == std::this_thread::get_id());
    }

    // 将任务压入队列
    void PushQueue(Functor cb)
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.push_back(cb);
        }

        WeakUpEventFd();
    }

    // 创建eventfd
    static int CreateEventfd()
    {
        int event_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (event_fd < 0)
        {
            LOG(LogLevel::ERROR) << " eventfd create error ...";
            abort();
        }

        return event_fd;
    }

    // 读取eventfd
    void ReadEventfd()
    {
        uint64_t val = 0;
        int n = read(_event_fd, &val, sizeof(val));
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return;
            }
            LOG(LogLevel::ERROR) << " eventfd read error ...";
            abort();
        }
    }

    // 唤醒线程
    void WeakUpEventFd()
    {
        uint64_t val = 1;
        int n = write(_event_fd, &val, sizeof(val));
        if (n < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            LOG(LogLevel::ERROR) << " eventfd write error ...";
            abort();
        }
    }

    // 开始循环监控执行任务
    void Start()
    {
        while (1)
        {
            // 1. 开始监控
            std::vector<Channel *> res;
            _poller.Poll(&res);

            // 2. 事件处理
            for (auto &e : res)
            {
                e->EventHandle();
            }

            // 3. 执行全部任务
            ExecuteTaskAll();
        }
    }

    // 执行任务
    void RuninLoop(Functor cb)
    {
        if (IsEventLoop())
        {
            cb();
            return;
        }

        PushQueue(cb);
    }

    // 修改或者添加监控
    void UpdataEvent(Channel *channel)
    {
        _poller.UpdataEvent(channel);
    }

    // 删除监控
    void RemoveEvent(Channel *channel)
    {
        _poller.RemoveEvent(channel);
    }

    //添加时钟任务
    void TimerAdd(uint64_t id, uint32_t interval, TaskCallback tc)
    {
        _timer_wheel.TimerAdd(id, interval, tc);
    }

    //刷新时钟任务
    void RefreshTask(uint64_t id)
    {
        _timer_wheel.RefreshTask(id);
    }

    //下线时钟任务
    void ShutTaskClock(uint64_t id)
    {
        _timer_wheel.ShutTaskClock(id);
    }

    //是否存在定时任务
    bool IsTimer(uint64_t id)
    {
        return _timer_wheel.IsTimer(id);
    }

private:
    int _event_fd;
    Poller _poller;
    std::unique_ptr<Channel> _channel;
    std::vector<Functor> _tasks;
    std::mutex _mutex;
    std::thread::id _thread_id;
    TimerWheel _timer_wheel;
};


//一个EventLoop线程
class LoopThread
{
public:
    LoopThread()
        : _loop(nullptr),
          _thread(&LoopThread::Headler, this)
    {
    }

    //获取当前线程中唯一的EventLoop
    EventLoop *GetLoop()
    {
        //需要使用锁来判定是否构建完EventLoop了
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]()
                       { return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }

private:
    //构建EventLoop执行Start
    void Headler()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }

        loop.Start();
    }

private:
    EventLoop *_loop;
    std::thread _thread;
    std::mutex _mutex;
    std::condition_variable _cond;
};


//EventLoop线程池
class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop *base_loop)
        : _thread_count(0),
          _base_loop(base_loop),
          _loop_index(0)
    {
    }

    //设置线程个数
    void SetThreadCount(int thread_count)
    {
        _thread_count = thread_count;
    }

    //创建线程
    void Create()
    {
        if (_thread_count > 0)
        {
            _loop_threads.resize(_thread_count);
            _event_loops.resize(_thread_count);

            for (int i = 0; i < _thread_count; ++i)
            {
                _loop_threads[i] = new LoopThread();
                _event_loops[i] = _loop_threads[i]->GetLoop();
            }
        }
    }

    //获取一个线程 -- 采用RR轮转的方式
    EventLoop *GetEventLoop()
    {
        if (_thread_count == 0)
        {
            return _base_loop;
        }

        int index = _loop_index % _thread_count;
        ++_loop_index;
        return _event_loops[index];
    }

private:
    int _thread_count;
    int _loop_index;
    EventLoop *_base_loop;
    std::vector<LoopThread *> _loop_threads;
    std::vector<EventLoop *> _event_loops;
};


//通用类型
class Any
{
public:
    Any()
        : _holder(nullptr)
    {
    }

    template <class T>
    Any(const T &val)
        : _holder(new placeholder<T>(val))
    {
    }

    Any(const Any &val)
        : _holder(val._holder->clone())
    {
    }

    ~Any()
    {
        if (nullptr != _holder)
            delete _holder;
    }

    template <class T>
    Any &operator=(const T &val)
    {
        Any(val).swap(*this);
        return *this;
    }
    Any &operator=(const Any &val)
    {
        Any(val).swap(*this);
        return *this;
    }

    template <class T>
    T *get()
    {
        assert(nullptr != _holder);
        assert(_holder->type() == typeid(T));
        return &(((placeholder<T> *)_holder)->_val);
    }

    void swap(Any &val)
    {
        std::swap(this->_holder, val._holder);
    }

private:
    class holder
    {
    public:
        virtual ~holder() {};
        holder() = default;

        virtual const std::type_info &type() = 0;
        virtual holder *clone() = 0;
    };

    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder() = default;
        placeholder(const T &val)
            : _val(val)
        {
        }

        virtual const std::type_info &type()
        {
            return typeid(T);
        }
        virtual holder *clone()
        {
            return new placeholder(this->_val);
        }
        ~placeholder() = default;

    public:
        T _val;
    };

private:
    holder *_holder;
};


class Connection;
using ConnectionPtr = std::shared_ptr<Connection>;
using ConnectedCallback = std::function<void(const ConnectionPtr &)>;
using MessageCallback = std::function<void(const ConnectionPtr &, Buffer *)>;
using ClosedCallback = std::function<void(const ConnectionPtr &)>;
using AnyEventCallback = std::function<void(const ConnectionPtr &)>;

// DISCONECTED -- 连接关闭状态；   CONNECTING -- 连接建立成功-待处理状态
// CONNECTED -- 连接建立完成，各种设置已完成，可以通信的状态；  DISCONNECTING -- 待关闭状态
typedef enum
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
} ConnStatus;

const size_t READSIZE = 65536;

using AcceptCallback = std::function<void(int)>;
//获取连接
class Accept
{
public:
    Accept(EventLoop *loop, uint16_t port)
        : _loop(loop),
          _sock(CreateAccept(port),true),//XZY
          _channel(_sock.Fd(), _loop)
    {
        _channel.SetReadCallback(std::bind(&Accept::ReadAeeept, this));
    }

    void SetAcceptCallback(const AcceptCallback &accept_cb)
    {
        _accept_cb = accept_cb;
    }

    //开始监听
    void Listen()
    {
        _channel.EnableRead();
    }

private:
    //创建一个监听套接字
    int CreateAccept(uint16_t port)
    {
        bool ret = _sock.CreateServer("0.0.0.0", port);
        if (ret == false)
        {
            abort();
        }

        return _sock.Fd();
    }

    //获取新连接
    void ReadAeeept() 
    {
        while(1) //XZY
        {
            int newfd = _sock.Accept();
            if (newfd == -2)
            {
                 continue;
            }
            else if(newfd == -1 || newfd == -3)
            {
                return;
            }

            if(_accept_cb)
                _accept_cb(newfd);
        }
        
    }

private:
    Socket _sock;
    EventLoop *_loop;
    Channel _channel;
    AcceptCallback _accept_cb;
};

/*要用shared_from_this需要继承std::enable_shared_from_this*/
class Connection : public std::enable_shared_from_this<Connection>
{
public:
    Connection(uint64_t conn_id, int sockfd, EventLoop *loop)
        : _conn_id(conn_id),
          _sockfd(sockfd),
          _loop(loop),
          _sock(_sockfd,true),//XZY
          _channel(_sockfd, _loop),
          _conn_status(ConnStatus::CONNECTING),
          _overtime_switch(false)
    {
        _channel.SetReadCallback(std::bind(&Connection::HandlerRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandlerWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandlerClose, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandlerError, this));
        _channel.SetEventCallback(std::bind(&Connection::HandlerEvent, this));
    }
    // 获取socketfd
    int Fd()
    {
        return _sockfd;
    }

    // 获取id
    uint64_t Id()
    {
        return _conn_id;
    }

    // 获取连接状态
    bool ConnectionStatus()
    {
        return _conn_status == ConnStatus::CONNECTED;
    }

    // 是否启用了超时管理
    bool OvertimeSwitch()
    {
        return _overtime_switch;
    }

    // 连接建立就绪后，进行channel回调设置，启动读监控，调用_connected_callback
    void Established()
    {
        _loop->RuninLoop(std::bind(&Connection::EstablishedInLoop, this)); //111
    }

    // 发送数据
    void Send(const char *data, size_t len)
    {
        Buffer buf;
        buf.WritePush(data, len);
        _loop->RuninLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }

    // 设置协议上下文
    void SetContext(const Any &context)
    {
        _context = context;
    }

    // 获取协议上下文
    Any *GetContext()
    {
        return &_context;
    }

    // 设置相关回调函数
    void SetConnectedCallback(const ConnectedCallback &conn_cb)
    {
        _conn_cb = conn_cb;
    }
    void SetMessageCallback(const MessageCallback &msg_cb)
    {
        _msg_cb = msg_cb;
    }
    void SetClosedCallback(const ClosedCallback &close_cb)
    {
        _close_cb = close_cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &ev_cb)
    {
        _ev_cb = ev_cb;
    }
    void SetServerClosedCallback(const ClosedCallback &ser_close_cb)
    {
        _ser_close_cb = ser_close_cb;
    }

    // 启用超时管理
    void EnableInactiveRelease(int sec)
    {
        _loop->RuninLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }

    // 取消超时管理
    void CancelInactiveRelease()
    {
        _loop->RuninLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }

    // 连接关闭(半关闭)
    void ShutDwon()
    {
        _loop->RuninLoop(std::bind(&Connection::ShutDwonInLoop, this));
    }

    // 连接释放
    void Release()
    {
        _loop->PushQueue(std::bind(&Connection::ReleaseInLoop, this)); //111
    }

    // 切换协议---重置上下文以及阶段性回调处理函数 -- 而是这个接口必须在EventLoop线程中立即执行
    // 防备新的事件触发后，处理的时候，切换任务还没有被执行--会导致数据使用原协议处理了。
    void Upgrade(const Any &context, const ConnectedCallback &conn_cb, const MessageCallback &msg_cb,
                 const ClosedCallback &close_cb, const AnyEventCallback &ev_cb)
    {
        _loop->RuninLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn_cb, msg_cb, close_cb, ev_cb));
    }

    ~Connection() = default;

private:
    // 读事件
    void HandlerRead()
    {
        // 1.读取数据
        while(1)    //XZY
        {
            char buff[READSIZE] = {0};
            ssize_t ret = _sock.NotBlockRecv(buff, READSIZE - 1);
            if (ret == -1)
            {
                // 关闭连接
                ShutDwonInLoop();
                return;
            }
            else if(ret == -2)
            {
                continue;
            }
            else if (ret == 0)
            {
                break;
            }

            // 2.放入缓冲区
            _in_buffer.WritePush(buff, ret);
        }
       
        // 3.执行回调函数
        if (_in_buffer.ReadSize() > 0)
        {
            if (_msg_cb)
            {
                _msg_cb(shared_from_this(), &_in_buffer);
            }
        }
    }

    // 写事件
    void HandlerWrite()
    {
        // 1.发送数据
        while(1)    //XZY
        {
            ssize_t ret = _sock.Send(_out_buffer.ReadPtr(), _out_buffer.ReadSize());
            if (ret == -1)
            {
                if (_in_buffer.ReadSize() > 0)
                {
                    if (_msg_cb)
                    {
                        _msg_cb(shared_from_this(), &_in_buffer);
                    }
                    // 释放连接
                    Release();
                    return;
                }
                break;
            }
            else if(ret == -2)
            {
                continue;
            }
            else if(ret == 0 || _out_buffer.ReadSize() == 0)
            {
                break;
            }
            LOG(LogLevel::DEBUG) << "ret : " <<ret;
            // 2.移动
            _out_buffer.ReadIndexSkewing(ret);
        }
       

        // 3.关闭可写事件监控
        if (_out_buffer.ReadSize() == 0)
        {
            _channel.RelieveWrite();
            if (_conn_status == DISCONNECTING)
            {
                // 释放连接
                Release();
            }
        }
    }

    // 关闭事件
    void HandlerClose()
    {
        if (_in_buffer.ReadSize() > 0)
        {
            if (_msg_cb)
            {
                _msg_cb(shared_from_this(), &_in_buffer);
            }
        }

        Release();
    }

    // 错误事件
    void HandlerError()
    {
        HandlerClose();
    }

    // 任意事件
    void HandlerEvent()
    {
        // 1.刷新任务
        if (_overtime_switch)
        {
            _loop->RefreshTask(_conn_id);
        }
        // 2.调用任意事件
        if (_ev_cb)
        {
            _ev_cb(shared_from_this());
        }
    }

    //完成创建连接
    void EstablishedInLoop()
    {
        assert(_conn_status = ConnStatus::CONNECTING);
        // 1.改变连接状态
        _conn_status = ConnStatus::CONNECTED;

         // 3.启用可读事件
         _channel.EnableRead();

        // 2.调用连接函数
        if (_conn_cb)
        {
            _conn_cb(shared_from_this());
        }

       
    }

    //发送数据
    void SendInLoop(Buffer& in)
    {
        if (_conn_status == ConnStatus::DISCONNECTED)
        {
            return;
        }

        // 1.写入获取区
        _out_buffer.WriteBufferPush(in);

        // 2.启动写事件
        if (_channel.IsMonitorWrite() == false)
        {
            _channel.EnableWrite();
        }
    }

    //刷新或者添加定时任务
    void EnableInactiveReleaseInLoop(int sec)
    {
        _overtime_switch = true;
        // 1.不存在 加入任务
        if (_loop->IsTimer(_conn_id) == false)
        {
            LOG(LogLevel::DEBUG)<<"不存在 加入任务 : " << sec;
            _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::HandlerClose, this));
            return;
        }

        // 2.存在刷新任务
        _loop->RefreshTask(_conn_id);
    }

    //下线定时任务
    void CancelInactiveReleaseInLoop()
    {
        _overtime_switch = false;
        if (_loop->IsTimer(_conn_id))
        {
            _loop->ShutTaskClock(_conn_id);
        }
    }

    //下线连接
    void ShutDwonInLoop()
    {
        _conn_status = ConnStatus::DISCONNECTING;

        if (_in_buffer.ReadSize() > 0)
        {
            if (_msg_cb)
            {
                _msg_cb(shared_from_this(), &_in_buffer);
            }
        }

        if (_out_buffer.ReadSize() > 0)
        {
            if (_channel.IsMonitorWrite() == false)
            {
                _channel.RelieveWrite();
            }
        }
        
        if(_out_buffer.ReadSize() == 0) //1111
        {
            Release();
        }
    }

    //删除连接
    void ReleaseInLoop()
    {
        if(_conn_status == ConnStatus::DISCONNECTED)
        {
            return;
        }
        // 更改状态
        _conn_status = ConnStatus::DISCONNECTED;

        // 移除监控
        _channel.Remove();

        // 关闭描述符
        _sock.Close();

        // 删除定时
        if (_loop->IsTimer(_conn_id))
        {
            CancelInactiveReleaseInLoop();
        }
      
  
        // 调用相关函数
        if (_close_cb)
        {
            _close_cb(shared_from_this());
        }

        if (_ser_close_cb)
        {
            _ser_close_cb(shared_from_this());
        }

        LOG(LogLevel::INFO) << "结束连接";
    }
    
    void UpgradeInLoop(const Any &context, const ConnectedCallback &conn_cb, const MessageCallback &msg_cb,
                       const ClosedCallback &close_cb, const AnyEventCallback &ev_cb)
    {
        if (_loop->IsEventLoop())
        {
            LOG(LogLevel::ERROR) << "UpgradeInLoop 不在loop线程工作 ...";
            abort();
        }
        _context = context;
        _conn_cb = conn_cb;
        _msg_cb = msg_cb;
        _close_cb = close_cb;
        _ev_cb = ev_cb;
    }

private:
    uint64_t _conn_id;       // 连接id
    int _sockfd;             // 连接socketfd
    ConnStatus _conn_status; // 连接状态
    Socket _sock;            // 连接socket管理
    bool _overtime_switch;   // 是否开启超时管理
    EventLoop *_loop;        // 事件监控管理
    Buffer _in_buffer;       // 输入缓冲区
    Buffer _out_buffer;      // 输出缓冲区
    Channel _channel;        // 事件回调管理
    Any _context;            // 连接上下文件管理

    // 相关回调函数
    ConnectedCallback _conn_cb;   // 连接成功回调
    MessageCallback _msg_cb;      // 消息处理回调
    ClosedCallback _close_cb;     // 连接关闭回调
    AnyEventCallback _ev_cb;      // 任意事件回调
    ClosedCallback _ser_close_cb; // 给服务器组件清理管理连接使用
};

class TcpServer
{
public:
    TcpServer(uint16_t port)
        : _next_id(0),
          _accept(&_base_loop, port),
          _loop_thread_pool(&_base_loop),
          _open_timeout(false)
    {
        _accept.SetAcceptCallback(std::bind(&TcpServer::NewConn, this, std::placeholders::_1));
        _accept.Listen();
    }

    // 设置从线程个数
    void SetThreadCount(int count)
    {
        _loop_thread_pool.SetThreadCount(count);
    }

    // 开始超时时间
    void OpenTimeout(int timeout)
    {
        _timeout = timeout;
        _open_timeout = true;
    }

    // 设置相关回调函数
    void SetConnectedCallback(const ConnectedCallback &conn_cb)
    {
        _conn_cb = conn_cb;
    }
    void SetMessageCallback(const MessageCallback &msg_cb)
    {
        _msg_cb = msg_cb;
    }
    void SetClosedCallback(const ClosedCallback &close_cb)
    {
        _close_cb = close_cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &ev_cb)
    {
        _ev_cb = ev_cb;
    }

    // 设置定时任务
    void RunAfter(const TaskCallback &task, int delay)
    {
        _base_loop.RuninLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }

    //开始
    void Start()
    {
        _loop_thread_pool.Create();
        _base_loop.Start();
    }

private:
    //刷新定时任务
    void RunAfterInLoop(const TaskCallback &task, int delay)
    {
        _next_id++;
        _base_loop.TimerAdd(_next_id, delay, task);
    }

    //创建一个新Connection
    void NewConn(int new_sockfd)
    {
        ConnectionPtr conn(new Connection(_next_id, new_sockfd, _loop_thread_pool.GetEventLoop()));

        conn->SetMessageCallback(_msg_cb);
        conn->SetClosedCallback(_close_cb);
        conn->SetConnectedCallback(_conn_cb);
        conn->SetAnyEventCallback(_ev_cb);
        conn->SetServerClosedCallback(std::bind(&TcpServer::RemoveConn, this, std::placeholders::_1));

        if (_open_timeout)
        {
            LOG(LogLevel::DEBUG)<<"启动非活跃超时销毁 : " << _timeout;
            conn->EnableInactiveRelease(_timeout); // 启动非活跃超时销毁
        }

        conn->Established();                           // 就绪初始化
        _conns.insert(std::make_pair(_next_id, conn)); // 插入管理

        ++_next_id;
    }

    // 移除管理
    void RemoveConn(const ConnectionPtr &conn)
    {
        _base_loop.RuninLoop(std::bind(&TcpServer::RemoveConnInLoop, this, conn));
    }


    void RemoveConnInLoop(const ConnectionPtr &conn)
    {
        uint64_t id = conn->Id();
        auto it = _conns.find(id);
        if (it == _conns.end())
        {
            return;
        }

        _conns.erase(it);
    }

private:
    uint64_t _next_id;                                  // 自增id
    EventLoop _base_loop;                               // 主 EventLoop
    Accept _accept;                                     // 监听套接字
    std::unordered_map<uint64_t, ConnectionPtr> _conns; // 管理连接
    LoopThreadPool _loop_thread_pool;                   // 从EventLoop
    int _timeout;                                       // 超时时间
    bool _open_timeout;                                 // 是否开启超时时间

    // 相关回调函数
    ConnectedCallback _conn_cb = nullptr; // 连接成功回调
    MessageCallback _msg_cb = nullptr;    // 消息处理回调
    ClosedCallback _close_cb = nullptr;   // 连接关闭回调
    AnyEventCallback _ev_cb = nullptr;    // 任意事件回调
};

//添加或者修改监控
void Channel::Updata()
{
    _loop->UpdataEvent(this);
}

// 删除监控
void Channel::Remove()
{
    _loop->RemoveEvent(this);
}

//添加定时任务
void TimerWheel::TimerAdd(uint64_t id, uint32_t interval, TaskCallback tc)
{
    _loop->RuninLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, interval, tc));
}

//刷新定时任务
void TimerWheel::RefreshTask(uint64_t id)
{
    _loop->RuninLoop(std::bind(&TimerWheel::RefreshTaskInLoop, this, id));
}

//下线一个定时任务
void TimerWheel::ShutTaskClock(uint64_t id)
{
    _loop->RuninLoop(std::bind(&TimerWheel::ShutTaskClockInLoop, this, id));
}


//忽略信号
class NetWork
{
public:
    NetWork()
    {
        signal(SIGPIPE, SIG_IGN);
    }
};
static NetWork nw;